package com.atguigu.realtime.app.dwd.db;

import com.atguigu.realtime.utils.KafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdTradeCartAdd {
    public static void main(String[] args) {
        //TODO 1 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //TODO 2 从Kafka读取业务数据 封装为flink表
        tableEnv.executeSql("create table topic_db(\n" +
                "`database` string,\n" +
                "`table` string,\n" +
                "`type` string,\n" +
                "`data` map<string,string>,\n" +
                "`old` map<string,string>,\n" +
                "`ts` string,\n" +
                "`pt` as PROCTIME()\n" +
                ")" + KafkaUtil.getKafkaDDL("topic_db","dwd_trade_cart_add"));

        //TODO 3读取购物车表数据
        Table cartAdd = tableEnv.sqlQuery("select\n" +
                "`data`['id'] id,\n" +
                "`data`['user_id'] user_id,\n" +
                "`data`['course_id'] course_id,\n" +
                "`data`['course_name'] course_name,\n" +
                "`data`['cart_price'] cart_price,\n" +
                "`data`['session_id'] session_id,\n" +
                "`data`['create_time'] create_time,\n" +
                "`data`['update_time'] update_time,\n" +
                "`data`['sold'] sold\n" +
                "from\n" +
                "`topic_db`\n" +
                "where `table` = 'cart_info'\n" +
                "and (`type` = 'insert'\n" +
                ")");
        tableEnv.createTemporaryView("cart_add",cartAdd);

        //TODO 4 获取加购明细表
        Table resultTable = tableEnv.sqlQuery("select\n" +
                "id,\n" +
                "user_id,\n" +
                "course_id,\n" +
                "course_name,\n" +
                "cart_price,\n" +
                "session_id,\n" +
                "create_time,\n" +
                "update_time,\n" +
                "sold\n" +
                "from\n" +
                "cart_add");
        tableEnv.createTemporaryView("result_table",resultTable);

        //TODO 5 建立 Kafka-Connector dwd_trade_cart_add 表
        tableEnv.executeSql("create table dwd_trade_cart_add(\n" +
                "id string,\n" +
                "user_id string,\n" +
                "course_id String,\n" +
                "course_name String,\n" +
                "cart_price String,\n" +
                "session_id String,\n" +
                "create_time String,\n" +
                "update_time String,\n" +
                "sold String\n" +
                ")" + KafkaUtil.getKafkaSinkDDL("dwd_trade_cart_add"));

        //TODO 6 将关联结果写入 Kafka-Connector 表
        tableEnv.executeSql("insert into dwd_trade_cart_add select * from result_table");
    }
}
